package com.ruyuan.seckilling.purchase.async.listener;

import com.ruyuan.seckilling.common.CacheKey;
import com.ruyuan.seckilling.purchase.async.BasePurchaseListener;
import com.ruyuan.seckilling.purchase.async.PurchaseContext;
import com.ruyuan.seckilling.purchase.async.event.Step3LockStock;
import com.ruyuan.seckilling.purchase.async.event.Step4CreateOrder;
import com.ruyuan.starter.async.anno.Channel;
import com.ruyuan.starter.async.event.BaseEvent;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;

import java.util.concurrent.atomic.AtomicLong;

/**
 * 扣减库存
 *
 * @author zhonghuashishan
 */
@Slf4j
@Component
@Channel("step3")
public class Step3Listener extends BasePurchaseListener<Step3LockStock> {

    private static final AtomicLong sequencer = new AtomicLong();
    private static final String SCRIPT = "local stockKey = '%s';"
                    + "local salableStock = redis.call('hget', stockKey, 'salableStock') + 0;"
                    + "local lockedStock = redis.call('hget', stockKey, 'lockedStock') + 0;"
                    + "if(salableStock > 0) "
                    + "then "
                    + "redis.call('hset', stockKey, 'salableStock', salableStock - 1);"
                    + "redis.call('hset', stockKey, 'lockedStock', lockedStock + 1);"
                    + "return 'success';"
                    + "else "
                    + "return 'failure';"
                    + "end;";

    @Override
    public boolean accept(BaseEvent event) {
        return event instanceof Step3LockStock;
    }

    @Override
    protected void doThisStep(Step3LockStock event, PurchaseContext purchaseContext) {
        executorService.execute("step3", () -> {
            // 首先获取一个自增的序列，在第1次扣减库存时用它来决定扣减哪一台redis上的库存
            long sequence = sequencer.incrementAndGet();

            Long activity = purchaseContext.getActivityId();
            Long userId = purchaseContext.getUserId();
            Long skuId = purchaseContext.getSkuId();

            String stockKey = CacheKey.buildStockKey(activity, skuId);
            String script = String.format(SCRIPT, stockKey);

            // 通过sequence定位到一台扣库存的起始的redis，大家都随机的选择一台redis开始扣库存
            // 如果在起始的redis上没有扣减库存成功，说明起始的redis上没有库存了，但是此时其他的redis上可能还有库存
            // 这种情况开始尝试在下一台redis上扣减库存
            int redisCount = cacheSupport.getRedisCount();
            long maxSequence = sequence + redisCount - 1;
            String result;
            for (long i = sequence; i <= maxSequence; i++) {
                log.info("扣减库存，sequence={}", i);
                result = (String) cacheSupport.eval(i, script);
                if (StringUtils.equals(result, "success")) {
                    // 扣减库存成功后把用户已经抢购成功记录到redis中
                    String key = CacheKey.buildCheckProductKey(activity, userId, skuId);
                    cacheSupport.set(key, "1");
                    cacheSupport.expire(key, 7200);
                    // 需要记录下是在哪台redis上扣的库存，后面确认库存的时候，还需要在这台redis上确认
                    purchaseContext.setSequence(i);

                    log.info("扣减库存，扣减库存成功，sequence={}", i);

                    // 抢购成功后，进入下一步发送创建订单的消息
                    bossEventBus.publish("step4", new Step4CreateOrder(), purchaseContext);
                    return;
                }
            }
            response(purchaseContext.getAsyncContext(), false, "该商品已经售罄了");
        });
    }
}
